package com.xctech.yace.service;

import com.xctech.yace.pool.ProduceThreadPool;
import com.xctech.yace.thread.ConsumerThread;
import com.xctech.yace.thread.ConsumerTpsThread;
import com.xctech.yace.util.DirUtil;
import lombok.Data;
import org.apache.kafka.clients.consumer.KafkaConsumer;


import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ThreadPoolExecutor;

import static cn.hutool.core.thread.GlobalThreadPool.execute;

@Data
public class KafkaRecived {

    private Properties props;

    private KafkaConsumer<String,String> consumer;

    public KafkaRecived() {
        String filePath = DirUtil.getOrgPath()+"conf/kafkaconsumer.properties";
        props = new Properties();
        try {
            props.load(new FileInputStream(filePath));
        } catch (IOException e) {
            e.printStackTrace();
        }
        this.consumer = new KafkaConsumer<String, String>(props);
    }

    /**
     * 开启接受消息
     * @param topics 主题列表
     * @return
     */
    public ConsumerThread reciveMsg(String[] topics){
        ConsumerThread consumerThread = new ConsumerThread(consumer, topics);
        ThreadPoolExecutor instance = ProduceThreadPool.getInstance();
        instance.execute(consumerThread);
        return consumerThread;
    }

    public ConsumerThread reciveConsumer(String[] topics){
        ConsumerThread consumerThread = new ConsumerThread(consumer, topics);
        return consumerThread;
    }

    /**
     * 开启接受消息tps
     * @param topics 主题列表
     * @return
     */
    public void reciveMsgTps(String[] topics){
        ProduceThreadPool.getInstance().execute(new ConsumerTpsThread(consumer,topics));
    }

    /**
     * 线程池停止
     * @return
     */
    public void stopReciveThreadPool(Integer second){
        ProduceThreadPool.stopPool(1000l*second);
    }

}
